package com.boot.study.api

import java.util
import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.functions.{RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.executiongraph.restart.RestartStrategy
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object StateTest {
  def main(args: Array[String]): Unit = {
    // 创建批处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行
    env.setParallelism(1)
    // 设置检查点 1秒触发一次 毫秒级别
    env.enableCheckpointing(1000L)
    // 只处理一次数据
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 超时时间 一般分钟级别
    env.getCheckpointConfig.setCheckpointTimeout(60000L)
    // 允许最大进行几个checkpoints同时保存，默认是单个
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
    // 两个checkpoint执行中的最小间隔时间 与上面的并行度冲突，两者只能取其一
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // 在多个故障点恢复时是否更倾向于checkpoint,默认是false,谁离得近取谁的值
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
    // 容忍checkpoint失败几次
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)

    // 重启策略 固定时间重启  参数1：重启次数  参数2：时间间隔
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000))
    // 失败率重启 参数1：重启次数 参数2：重启的时间范围 参数3：重启的时间间隔
    env.setRestartStrategy(RestartStrategies.failureRateRestart(5,Time.of(5,TimeUnit.MINUTES),Time.of(1,TimeUnit.SECONDS)))
    // 从外部命令中提取参数
    // windows启动 nc64.exe -L -p 9000
    // --host 127.0.0.1 --port 9000
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = parameterTool.get("host")
    val port: Int = parameterTool.getInt("port")

    // 接受socket文本流
    val dataStream: DataStream[String] = env.socketTextStream(host, port)
    // 转换处理分组
    val resultDataStream: DataStream[SensorReading] = dataStream.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    val alertStream: DataStream[(String, Double, Double)] = resultDataStream.keyBy(_.id).flatMap(new TempChangeAlert(10.0))
    // 对于温度传感器温度值跳变，超过10度就报警
    alertStream.print("state")
    // 启动任务执行
    env.execute("state test")
  }
}

// 实现自定义RichFlatMapFunction
class TempChangeAlert(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
  // 定义状态，保存上一次的温度值
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))

  override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    // 获取上次的温度值
    val lastTemp: Double = lastTempState.value()
    val diff: Double = (value.temperature - lastTemp).abs
    if (diff > threshold) {
      out.collect((value.id, lastTemp, value.temperature))
    }
    // 更新状态
    lastTempState.update(value.temperature)
  }
}

// keyed state测试：必须定义在RichFunction，因为需要运行时上下文
class MyRichMapper extends RichMapFunction[SensorReading, String] {

  var valueState: ValueState[Double] = _
  lazy val listState: ListState[Int] = getRuntimeContext.getListState(new ListStateDescriptor[Int]("listState", classOf[Int]))

  override def open(parameters: Configuration): Unit = {
    valueState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("valueState", classOf[Double]))
  }

  override def map(value: SensorReading): String = {
    // 状态的读写
    val v = valueState.value()
    valueState.update(value.temperature)

    listState.add(1)
    listState.addAll(new util.ArrayList[Int](util.Arrays.asList(1, 2, 3)))
    listState.update(new util.ArrayList[Int](util.Arrays.asList(4, 5, 6)))
    val ls: java.lang.Iterable[Int] = listState.get()

    value.id
  }
}